package com.lagou.rpc.consumer;

import com.alibaba.fastjson.JSON;
import com.lagou.rpc.api.IUserService;
import com.lagou.rpc.common.ServiceInfo;
import com.lagou.rpc.consumer.proxy.RpcClientProxy;
import com.lagou.rpc.pojo.User;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundPathable;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;

public class ClientBootStrap {

    private static CuratorFramework client;

    private static Map<String, ServiceInfo> rpcServicesMap = new HashMap<>();

    public static void main(String[] args) {

        client = CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181") //server地址
                .sessionTimeoutMs(5000) // 会话超时时间
                .connectionTimeoutMs(3000) // 连接超时时间
                .retryPolicy(new ExponentialBackoffRetry(1000,5)) // 重试策略
                .namespace("tuomaxu")
                .build(); //
        client.start();
        System.out.println("Zookeeper 连接成功");

        //监控子节点，获取服务上线的信息和服务响应的信息
        PathChildrenCache pathChildrenCache = new PathChildrenCache(client, "/provider", true);
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {

                if(event.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED){
                    String name = event.getData().getPath();
                    System.out.println("服务"+name+"上线");
                    ServiceInfo serviceInfo = JSON.parseObject(new String(event.getData().getData(), Charset.forName("UTF-8")), ServiceInfo.class);
                    RpcClientProxy.remoteServiceInfoMap.put(name,serviceInfo);
                }

                if(event.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED){
                    String name = event.getData().getPath();
                    System.out.println("服务"+name+"下线");
                    RpcClientProxy.remoteServiceInfoMap.remove(name);
                }

                if(event.getType() == PathChildrenCacheEvent.Type.CHILD_UPDATED){
                    String name = event.getData().getPath();
                    System.out.println("服务"+name+"变化");
                    ServiceInfo serviceInfo = JSON.parseObject(new String(event.getData().getData(), Charset.forName("UTF-8")), ServiceInfo.class);
                    RpcClientProxy.remoteServiceInfoMap.put(name,serviceInfo);
                }

            }
        });

        try {
            pathChildrenCache.start();
            System.out.println("开始监听");

            //开启一个子线程，每5秒上传本地服务调用信息
            new Thread(new Runnable() {
                @Override
                public void run() {
                    while (true){
                        try {
                            //每5秒提交一次
                            Thread.sleep(5000);

                            //提交本地缓存的5秒内的请求响应时间信息
                            Map<String, ServiceInfo> localServiceInfoMap = RpcClientProxy.localServiceInfoMap;
                            for (Map.Entry<String, ServiceInfo> entry : localServiceInfoMap.entrySet()) {
                                String name = entry.getKey();
                                ServiceInfo serviceInfo = entry.getValue();
                                //如果都为零，说明本次缓存窗口中没有请求，不提交
                                if(serviceInfo.getRequestTimestamp() == 0 && serviceInfo.getResponseDuration() == 0){
                                    continue;
                                }
                                //判断当前时间与最后一次请求时间之差是否大于5秒，如果大于，请求时间和响应时间置零
                                long currentTime = System.currentTimeMillis();
                                if(currentTime-serviceInfo.getRequestTimestamp() > 5000){
                                    serviceInfo.setResponseDuration(0);
                                    serviceInfo.setRequestTimestamp(0);
                                }
                                String data = JSON.toJSONString(serviceInfo);
                                client.setData().forPath(name,data.getBytes(StandardCharsets.UTF_8));
                                System.out.println("服务"+name+"提交成功");
                            }


                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            }).start();


            //循环模拟客户端调用
            while (true){
                Random ra =new Random();
                Thread.sleep((ra.nextInt(2)+1)*1000);

                IUserService userService = (IUserService)RpcClientProxy.createProxy(IUserService.class);
                if(userService != null){
                    User user = userService.getById(1);
                    System.out.println(user);
                }
            }


        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
